package edu.kit.aifb.ldstream.repository.event;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Semaphore;

import edu.kit.aifb.ldstream.env.Environment;

public class EventManager{

	private ConcurrentLinkedQueue<StreamEvent> queue;
	private EventWorker[] workers;
	private Semaphore semaphore;

	public EventManager(){
		this.semaphore = new Semaphore(0, true);
		this.workers = new EventWorker[Environment.NUM_THREADS];
		this.queue = new ConcurrentLinkedQueue<StreamEvent>();
		for(int i = 0; i < Environment.NUM_THREADS; i++){
			this.workers[i] = new EventWorker(this.queue, this.semaphore);
			this.workers[i].start();
		}
	}

	/**
	 * Adding an {@code StreamEvent} which will be save to the Repository.
	 * @param event {@code StreamEvent} which will be saved.
	 * @param baseURI baseURI of the given {@code StreamEvent}.
	 */
	public void addEvent(String event, String baseURI){
		this.queue.add(new StreamEvent(event, baseURI));
		// "increment" the semaphore. Permits of the semaphore are corresponding to the size of the queue.
		this.semaphore.release();
	}
	/**
	 * Adding an {@code StreamEvent} which will be save to the Repository.
	 * @param event {@code StreamEvent} which will be saved.
	 * @param baseURI baseURI of the given {@code StreamEvent}.
	 * @param URI additional URI which will be displayed on the console. 
	 */
	// the URI is used to print the success of saving that URI to the console.
	public void addEvent(String event, String baseURI, String URI){
		this.queue.add(new StreamEvent(event, baseURI, URI));
		// "increment" the semaphore. Permits of the semaphore are corresponding to the size of the queue.
		this.semaphore.release();
	}

	/**
	 * Signals the EventManager to shut down the EventWorkers. The Workers will finish their work first.
	 */
	public void shutDown(){
		while(this.semaphore.availablePermits() > 0){
			System.out.println("Permits left: " + this.semaphore.availablePermits());
			try {
				Thread.sleep(500);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		forceShutDown();
	}

	/**
	 * Forces the EventManager and the EventWorkers to shut down without finishing their work.
	 */
	public void forceShutDown(){
		for(EventWorker worker : this.workers){
			worker.isFinished();
		}
		// granting permits to the semaphore so every thread is able to check it's finish flag.
		this.semaphore.release(this.workers.length);
		for(EventWorker worker : this.workers){
			if(worker.isAlive()){
				try {
					worker.join();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
		this.semaphore.drainPermits();
	}

}
